package com.ruoyi.common.service;

import com.ruoyi.common.config.MqttConfig;
import com.ruoyi.common.config.MqttConnect;
import com.ruoyi.common.config.MyMqttCallback;
import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.mqttv5.client.MqttClient;
import org.eclipse.paho.mqttv5.client.MqttConnectionOptions;
import org.eclipse.paho.mqttv5.client.MqttToken;
import org.eclipse.paho.mqttv5.client.MqttTopic;
import org.eclipse.paho.mqttv5.client.persist.MemoryPersistence;
import org.eclipse.paho.mqttv5.common.MqttException;
import org.eclipse.paho.mqttv5.common.MqttMessage;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;

/**
 * 发布端：主要实现发布消息和订阅主题，接收消息在回调类PushCallback中
 * 要发布消息的时候只需要调用sendMQTTMessage方法就OK了
 */
@Slf4j
@Component
public class MqttServer {

    /* 订阅者客户端对象 */
    private MqttClient mqttClient;

    @Resource
    private MqttConnect mqttConnect;

    @Resource
    private MqttConfig config;

    public MqttServer() {
//        log.info("8081上线了");
    }

    /**
     * 发布者客户端和服务端建立连接
     */
    public MqttClient connectClient() {
        //防止重复创建MQTTClient实例
        try {
            if (mqttClient == null) {
                //先让客户端和服务器建立连接，MemoryPersistence设置clientid的保存形式，默认为以内存保存
                mqttClient = new MqttClient(config.getHost(), config.getClientId(), new MemoryPersistence());
                //发布消息不需要回调连接
                mqttClient.setCallback(new MyMqttCallback(config.getClientId()));
                log.info("client 连接成功！");
            }
            MqttConnectionOptions options = mqttConnect.getOptions();
            //判断拦截状态，这里注意一下，如果没有这个判断，是非常坑的
            if (!mqttClient.isConnected()) {
                mqttClient.connect(options);
            } else {//这里的逻辑是如果连接成功就重新连接
                mqttClient.disconnect();
                mqttClient.connect(mqttConnect.getOptions(options));
            }
        } catch (MqttException e) {
            log.info(e.toString());
        }
        return mqttClient;
    }

    /**
     * 消息发布(生产)
     *
     * @param topic   订阅主题
     * @param message 消息内容
     */
    private void publish(MqttTopic topic, MqttMessage message) {
        MqttToken token = null;
        try {
            //把消息发送给对应的主题
            token = topic.publish(message);
            token.waitForCompletion();
            //检查发送是否成功
            boolean flag = token.isComplete();

            StringBuffer sbf = new StringBuffer(200);
            sbf.append("topic" + topic.getName());
            sbf.append(", message：" + new String(message.getPayload()));
            if (flag) {
                sbf.append(" 成功！");
            } else {
                sbf.append(" 失败！");
            }
            log.info(sbf.toString());
        } catch (MqttException e) {
            log.info(e.toString());
        }
    }

    /**
     * 消息发布(生产)
     *
     * @param topic 主题
     * @param data  消息内容
     * @param qos   消息级别
     */
    public void sendMessage(String topic, String data, int qos) {
        try {
//            this.mqttClient = this.connectClient();
            /* 主题对象 */
            MqttTopic topic1 = this.mqttClient.getTopic(topic);
            /* 消息内容对象 */
            MqttMessage message = new MqttMessage();
            //消息等级
            //level 0：消息最多传递一次，不再关心它有没有发送到对方，也不设置任何重发机制
            //level 1：包含了简单的重发机制，发送消息之后等待接收者的回复，如果没收到回复则重新发送消息。这种模式能保证消息至少能到达一次，但无法保证消息重复
            //level 2： 有了重发和重复消息发现机制，保证消息到达对方并且严格只到达一次
            message.setQos(qos);
            //如果重复消费，则把值改为true,然后发送一条空的消息，之前的消息就会覆盖，然后在改为false
            message.setRetained(false);
            message.setPayload(data.getBytes());
            //将组装好的消息发出去
            this.publish(topic1, message);
        } catch (Exception e) {
            log.info(e.toString());
        }

    }

    /**
     * 消息订阅(消费)
     *
     * @param topic 要订阅的主题
     * @param qos   订阅消息的级别
     */
    public void subMessage(String topic, int qos) {
        try {
            //建立连接
            this.connectClient();
            //以某个消息级别订阅某个主题
            mqttClient.subscribe(topic, qos);
        } catch (MqttException e) {
            log.info(e.getMessage(), e);
        }
    }

    /**
     * 取消订阅
     *
     * @param topic 要订阅的主题
     */
    public void uniMessage(String topic) {
        try {
            //建立连接
            this.connectClient();
            //取消订阅某个主题
            //MQTT 协议中订阅关系是持久化的，因此如果不需要订阅某些 Topic，需要调用 unsubscribe 方法取消订阅关系。
            mqttClient.unsubscribe(topic);
        } catch (MqttException e) {
            log.info(e.getMessage(), e);
        }
    }

}